package cn.iocoder.yudao.framework.mqtt.listener;
import cn.iocoder.yudao.framework.mqtt.config.YudaoMqttConfiguration;
import cn.iocoder.yudao.framework.mqtt.core.annotation.MqttTopicListener;
import lombok.AllArgsConstructor;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.Arrays;

@Configuration
@AllArgsConstructor
public class MqttListenerRegister {

    private ApplicationContext context;

    @PostConstruct
    public void registerMqttListeners() {
        YudaoMqttConfiguration mqttConfiguration = context.getBean(YudaoMqttConfiguration.class);
        Arrays.stream(context.getBeansWithAnnotation(Component.class).values().toArray())
                .forEach(bean -> {
                    Method[] methods = bean.getClass().getDeclaredMethods();
                    for (Method method : methods) {
                        if (method.isAnnotationPresent(MqttTopicListener.class)) {
                            MqttTopicListener listener = method.getAnnotation(MqttTopicListener.class);
                            subscribeToTopic(listener.value(), listener.qos(), mqttConfiguration, bean, method);
                        }
                    }
                });
    }

    private void subscribeToTopic(String topic, int qos, YudaoMqttConfiguration mqttConfiguration, Object bean, Method method) {
        // 创建适配器
        IMqttMessageListener listener = (mqttTopic, message) -> {
            Object targetObject = context.getBean(bean.getClass());
            method.invoke(targetObject, mqttTopic, message);
        };

        try {
            mqttConfiguration.getMqttClient().subscribe(topic, qos, listener);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
